package com.lz.flow.usage;

import java.util.Arrays;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.function.Function;

/**
 * @Classname: com.lz.flow.usage.cx
 * @Description: TODO
 *
 * @Date: 2019/1/8 11:05
 * @Created by: li
 * @Version: V1.0
 */
public class MyFilterProcessorBootstrap {

    public static class MyFilterProcessor<T, K extends T> extends SubmissionPublisher<K> implements Flow
            .Processor<T, K> {

        private Function<? super T, Boolean> function;
        private Flow.Subscription subscription;

        public MyFilterProcessor(Function<? super T, Boolean> function) {
            super();
            this.function = function;
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            subscription.request(1);
        }

        @Override
        public void onNext(T item) {
            if (!function.apply(item)) {

                submit((K) item); // Blocking 阻塞保证所有接收
                // offer() // 根据返回是否true 或false 是否重试
            }

            subscription.request(1);
        }

        @Override
        public void onError(Throwable t) {
            t.printStackTrace();
        }

        @Override
        public void onComplete() {
            close();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        //Create Publisher
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //Create Processor and Subscriber
        MyFilterProcessor<String, String> filterProcessor = new MyFilterProcessor<>(s -> s.equals("x"));
        //创建的订阅者
        MySubscriber<String> subscriber = new MySubscriber<>();
        //Chain Processor and Subscriber
        publisher.subscribe(filterProcessor);
        filterProcessor.subscribe(subscriber);
        //发布Stream
        System.out.println("Publishing Items...");
        String[] items = {"1", "x", "2", "x", "3", "x"};
        Arrays.stream(items).forEach(i -> publisher.submit(i));

        publisher.close();
        Thread.sleep(20000);
    }
}
